﻿/*
* Copyright (C) 2020 ~ 2024 chaigec All Rights Reserved.
*
* Author:     chaigec <chaigec@163.com>
* Maintainer: chaigec <chaigec@163.com>

 * Use of this source code is governed by MIT license that can be found in the
 * LICENSE file in the root of the source tree. All contributing project authors
 * may be found in the AUTHORS file in the root of the source tree.
*/

#include "MkEvent.h"
#include "MkPipe.h"
//#include "MkNetworkUtil.h"
#include "MkUtil/MkThread.h"
#include "MkUtil/MkLog.h"
#include "MkUtil/MkUtil.h"

#ifndef _WIN32
#define MaxEpollSize      1024
#endif

CMkEvent::CMkEvent()
    : m_pPip(nullptr)
    , m_bRun(FALSE)
    , m_DispatchThreadId(0)
    , m_HasRemoved(FALSE)
    , m_Score(MaxInt32)
{
#ifndef _WIN32
    m_EpollFd = INVALID_SOCKET;
    m_pEvents = (struct epoll_event*)malloc(sizeof(struct epoll_event)*MaxEpollSize);
#endif // !_WIN32
}

Uint32 CMkEvent::Start()
{
#ifndef _WIN32
    m_EpollFd = epoll_create(MaxEpollSize);
    if (INVALID_SOCKET == m_EpollFd) {
        return SocketCreateFailed;
    }
#endif
    m_pPip = new CMkPipe(this);
    if (!m_pPip) {
        MkErrorLog("new pip failed\n");
        return MallocFailed;
    }
    m_bRun = TRUE;
    Uint32 ErrorCode = m_pPip->Start([this]() {
        MkList<PipTaskCallback> TmpTaskList;
        m_PipTaskLock.WLock();
        TmpTaskList = m_ListPipTask;
        m_ListPipTask.clear();
        m_PipTaskLock.WUnlock();
        MkList<PipTaskCallback>::iterator it = TmpTaskList.begin();
        for (; it != TmpTaskList.end(); it++) {
            if (*it) {
                (*it)();
            }
        }
    });
    MkCheckErrorLog(ErrorCode, "start pip failed\n");
    return m_EventThread.Start([this]() {
        DoLoop();
    });
}

Uint32 CMkEvent::Stop()
{
    m_bRun = FALSE;
    if (m_pPip) {
        m_pPip->Active();
    }
    m_EventThread.Stop();

    m_PipTaskLock.WLock();
    m_ListPipTask.clear();
    m_PipTaskLock.WUnlock();
    //include pip
#ifndef _WIN32
    MapEventer::iterator it = m_MapEventer.begin();
    for (; it != m_MapEventer.end(); it++) {
        if (it->first) {
            epoll_ctl(m_EpollFd, EPOLL_CTL_DEL, it->first, nullptr);
        }
    }
#endif // !_WIN32
    m_MapEventer.clear();
#ifndef _WIN32
    if (-1 != m_EpollFd) {
        close(m_EpollFd);
        m_EpollFd = -1;
    }
    MkFree(m_pEvents);
#endif
    MkDelete(m_pPip);
    return NoneError;
}

Uint32 CMkEvent::AddEventer(SOCKET Sock, Uint32 Event, const MkEventCallback& EventCb)
{
    if (CMkUtil::GetLocalThreadId() != m_DispatchThreadId
        && 0 != m_DispatchThreadId /*event dispath not start*/) {

        return AddPipTask([this, Sock, Event, EventCb]() {
            AddEventer(Sock, Event, EventCb);
        });
    }
#ifndef _WIN32
    struct epoll_event eEvent;
    eEvent.events = MkEventToEpollEvent(Event);
    eEvent.data.fd = Sock;
    //eEvent.data.ptr = pSocket;
    int nRet = epoll_ctl(m_EpollFd, EPOLL_CTL_ADD, Sock, &eEvent);
    if (0 != nRet) {
        MkErrorLog("add event failed socket:%d  this score:%d size:%d\n", Sock, m_Score, m_MapEventer.size());
        return SystemError;
    }
#endif
    MkEventerParam Param;
    Param.Event = Event;
    Param.EventCb = EventCb;
    m_MapEventer[Sock] = Param;
    m_Score--;
    return NoneError;
}

Uint32 CMkEvent::DeleteEventer(SOCKET Sock)
{
    MapEventer::iterator it = m_MapEventer.find(Sock);
    if (it != m_MapEventer.end()) {
        it->second.Event = EventNone;  //set not recv event first
    }
    if (CMkUtil::GetLocalThreadId() != m_DispatchThreadId
        && 0 != m_DispatchThreadId /*event dispath not start*/) {
        return AddPipTask([this, Sock]() {
            DeleteEventer(Sock);
        });
    }

#ifndef _WIN32
    epoll_ctl(m_EpollFd, EPOLL_CTL_DEL, Sock, nullptr);
#endif
    it = m_MapEventer.find(Sock);
    if (it != m_MapEventer.end()) {
        it->second.EventCb(EventRemove);
        m_MapEventer.erase(it);
    }
    m_HasRemoved = TRUE;
    m_Score++;
    return NoneError;
}

Uint32 CMkEvent::ModifyEventer(SOCKET Sock, Uint32 Event)
{
    if (CMkUtil::GetLocalThreadId() != m_DispatchThreadId) {
        return AddPipTask([this, Sock, Event]() {
            ModifyEventer(Sock, Event);
        });
    }
    MapEventer::iterator it = m_MapEventer.find(Sock);
    if (it != m_MapEventer.end()) {
        it->second.Event = Event;
    } else {
        return InvalidParams;
    }
#ifndef _WIN32
    struct epoll_event eEvent = { 0 };
    eEvent.events = MkEventToEpollEvent(Event);
    eEvent.data.fd = Sock;
    epoll_ctl(m_EpollFd, EPOLL_CTL_MOD, Sock, &eEvent);
#endif 
    return NoneError;
}

#ifndef _WIN32
Uint32 CMkEvent::MkEventToEpollEvent(Uint32 Event)
{
    return ((Event&EventRead) ? EPOLLIN : 0) | ((Event&EventWrite) ? EPOLLOUT : 0) | ((Event&EventError) ? (EPOLLERR | EPOLLHUP) : 0) | ((Event&EventEt) ? EPOLLET : 0);
}
#endif // !_WIN32

Uint32 CMkEvent::AddPipTask(PipTaskCallback Cb, BOOL bFirst/* = FALSE*/)
{
    if (!m_pPip) {
        return SystemError;
    }
    m_PipTaskLock.WLock();
    if (bFirst) {
        m_ListPipTask.push_front(Cb);
    } else {
        m_ListPipTask.push_back(Cb);
    }
    m_PipTaskLock.WUnlock();
    return  m_pPip->Active();
}

void CMkEvent::DoLoop()
{
    m_DispatchThreadId = CMkUtil::GetLocalThreadId();
    while (m_bRun) {
#ifdef _WIN32
        fd_set SetRead;
        fd_set SetWrite;
        fd_set SetError;
        FD_ZERO(&SetRead);
        FD_ZERO(&SetWrite);
        FD_ZERO(&SetError);
        int MaxFd = 0;
        struct timeval tv;
        tv.tv_sec = 60 * 60; //default one hour
        tv.tv_usec = 0;
#ifdef _DEBUG
        Uint32 EventSize = m_MapEventer.size();
#endif
        MapEventer::iterator it = m_MapEventer.begin();
        for (; it != m_MapEventer.end(); it++) {
            if (INVALID_SOCKET == it->first
                || !it->second.EventCb) {
                continue;
            }
            MaxFd = MkMax(MaxFd, static_cast<int>(it->first));
            if (it->second.Event & EventRead) {
                FD_SET(it->first, &SetRead);
            }
            if (it->second.Event & EventWrite) {
                FD_SET(it->first, &SetWrite);
            }
            if (it->second.Event & EventError) {
                FD_SET(it->first, &SetError);
            }
        }
        if (0 == MaxFd) {
            MaxFd = 1;
        }
        //CMkTime NowTime = NowMkTime;
        int ret = select(MaxFd + 100, &SetRead, &SetWrite, &SetError, &tv);
        //CMkTime EndTime = NowMkTime;
        //AddWaitMilliSec(NowTime, static_cast<Uint32>((EndTime - NowTime) / 1000));
        if (ret > 0) {
            it = m_MapEventer.begin();
            m_HasRemoved = FALSE;
            Uint32 Event = EventNone;
            for (; it != m_MapEventer.end(); it++) {
                Event = EventNone;
                if (FD_ISSET(it->first, &SetRead)) {
                    Event |= EventRead;
                } else if (FD_ISSET(it->first, &SetWrite)) {
                    Event |= EventWrite;
                } else if (FD_ISSET(it->first, &SetError)) {
                    Event |= EventError;
                }
                if (EventNone != Event) {
                    it->second.EventCb(Event);
                }
                if (m_HasRemoved) {
                    break;
                }
            }
        }
#else
        memset(m_pEvents, 0, sizeof(struct epoll_event)*MaxEpollSize);
        int nRet = epoll_wait(m_EpollFd, m_pEvents, MaxEpollSize, 60 * 60 * 1000);
        if (0 > nRet) {
            MkErrorLog("epoll:%d wait time:%d ret:%d error\n", m_EpollFd, 60 * 60 * 1000, nRet);
            continue;
        }
        m_HasRemoved = FALSE;
        for (int i = 0; i < nRet; i++) {
            MapEventer::iterator it = m_MapEventer.find(m_pEvents[i].data.fd);
            if (it == m_MapEventer.end() || !it->second.EventCb) {
                continue;
            }
            Uint32 Event = 0;
            if (m_pEvents[i].events&EPOLLIN) {
                Event |= EventRead;
            } else if (m_pEvents[i].events&EPOLLOUT) {
                Event |= EventWrite;
            } else if ((m_pEvents[i].events &EPOLLHUP)
                || (m_pEvents[i].events&EPOLLERR)) {
                Event |= EventError;
            }
            it->second.EventCb(Event);
            if (m_HasRemoved) {
                break;
            }
        }
#endif
    }
}

//Uint32 CMkEvent::GetScore()
//{
//    Uint32 Score = MaxUint32;
//    CMkTime NowTime = NowMkTime;
//    m_WaitTimeLock.RLock();
//    MkList<std::pair<CMkTime, Uint32>>::iterator it = m_WaitTimeList.begin();
//    for (; it != m_WaitTimeList.end(); it++) {
//        if (it->first.GetTvSec() + 1 == NowTime.GetTvSec()) {
//            Score += it->second;
//        }
//    }
//    m_WaitTimeLock.RUnlock();
//    Score -= m_MapEventer.size();
//    return Score;
//}
//
//void CMkEvent::AddWaitMilliSec(const CMkTime& NowTime, Uint32 WaitMillSec)
//{
//    m_WaitTimeLock.WLock();
//    MkList<std::pair<CMkTime, Uint32>>::iterator it = m_WaitTimeList.begin();
//    while (it != m_WaitTimeList.end()) {
//        if (NowTime.GetTvSec() > it->first.GetTvSec() + 2) {
//            it = m_WaitTimeList.erase(it);
//        } else {
//            it++;
//        }
//    }
//    m_WaitTimeList.push_back(std::pair<CMkTime, Uint32>(NowTime, WaitMillSec));
//    m_WaitTimeLock.WUnlock();
//}

CMkEvent::~CMkEvent()
{


}

